package org.budo.dubbo.protocol.async.repository.activemq;

import java.util.ArrayList;
import java.util.List;

import javax.jms.BytesMessage;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;

import org.budo.dubbo.protocol.async.repository.AsyncRepository;
import org.budo.dubbo.protocol.async.repository.BudoAsyncInvocation;
import org.budo.support.lang.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.RpcInvocation;

/**
 * @author lmw
 */
public class ActiveMQAsyncRepository extends AbstractActiveMQAsyncRepository {
    private static final Logger log = LoggerFactory.getLogger(ActiveMQAsyncRepository.class);

    private static final int RECEIVE_TIMEOUT = 100; // 毫秒

    /**
     * 引用端调用方法时发消息
     */
    @Override
    public void sendMessageWhenInvoke(Invocation invocation) {
        try {
            Class<?> interfaceType = invocation.getInvoker().getInterface();
            String destinationName = invocation.getInvoker().getUrl().getParameter(AsyncRepository.DESTINATION_NAME);
            if (StringUtil.isEmpty(destinationName)) {
                destinationName = "budo-async-" + interfaceType.getName();
            }

            BudoActiveMQSession activeMQSession = this.getActiveMQSessionOfInterface(interfaceType, destinationName);

            BudoAsyncInvocation budoAsyncInvocation = BudoAsyncInvocation.fromDubboInvocation(invocation);
            byte[] requestBody = this.serializeAsyncInvocation(budoAsyncInvocation, invocation.getInvoker());

            BytesMessage bytesMessage = activeMQSession.getSession().createBytesMessage();
            bytesMessage.writeBytes(requestBody);

            MessageProducer messageProducer = activeMQSession.producer();
            messageProducer.send(bytesMessage);

            if (log.isDebugEnabled()) {
                log.debug("#39 sendMessageWhenInvoke, bytesMessage=" + bytesMessage + ", invocation=" + invocation);
            }
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 提供端收到消息后调用方法
     * 
     * @see org.apache.activemq.ActiveMQMessageConsumer#receive(long)
     */
    @Override
    public Integer popMessageAndInvoke(Invoker invoker) {
        Class<?> interfaceType = invoker.getInterface();
        String destinationName = invoker.getUrl().getParameter(AsyncRepository.DESTINATION_NAME);
        if (StringUtil.isEmpty(destinationName)) {
            destinationName = "budo-async-" + interfaceType.getName();
        }

        List<BytesMessage> messages = new ArrayList<BytesMessage>();

        BudoActiveMQSession activeMQSession = null;
        try {
            activeMQSession = this.getActiveMQSessionOfInterface(interfaceType, destinationName);
            MessageConsumer messageConsumer = activeMQSession.consumer();

            int messageCount = 0;
            while (true) {
                BytesMessage bytesMessage = (BytesMessage) messageConsumer.receive(RECEIVE_TIMEOUT);
                if (null == bytesMessage) {
                    if (log.isDebugEnabled()) {
                        log.debug("#80 bytesMessage is null, destinationName=" + destinationName + ", messageConsumer=" + messageConsumer);
                    }

                    break; // 返回空就跳出
                }

                // 读取消息内容
                byte[] messageBody = new byte[(int) bytesMessage.getBodyLength()];
                bytesMessage.readBytes(messageBody);

                BudoAsyncInvocation budoAsyncInvocation = this.deserializeAsyncInvocation(messageBody, invoker);

                if (log.isDebugEnabled()) {
                    log.debug("#93 message received, bytesMessage=" + bytesMessage + ", budoAsyncInvocation=" + budoAsyncInvocation);
                }

                // 如果消息延迟就打日志
                this.logMessageDelay(budoAsyncInvocation);

                RpcInvocation invocation = BudoAsyncInvocation.toDubboInvocation(budoAsyncInvocation);
                invoker.invoke(invocation); // 执行方法

                // 暂不提交，finally 里提交
                messages.add(bytesMessage);
                messageCount++;
            }

            return messageCount;
        } catch (Throwable e) {
            return this.handleError(invoker, activeMQSession, e);
        } finally {
            this.acknowledgeMessages(messages); // 提交消息
        }
    }

    /**
     * 方法调用成功后确认收到消息，默认 ActiveMQSession.INDIVIDUAL_ACKNOWLEDG 单条确认模式
     * https://blog.csdn.net/zhu_tianwei/article/details/46303535
     */
    private void acknowledgeMessages(List<BytesMessage> messages) {
        for (BytesMessage message : messages) {
            try {
                message.acknowledge();
            } catch (JMSException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 出错时的处理，可能需要重连
     */
    private Integer handleError(Invoker invoker, BudoActiveMQSession activeMQSession, Throwable e) {
        if ((e + "").contains("The Consumer is closed")) {
            log.error("#107 popMessageAndInvoke error, invoker=" + invoker + ", e=" + e);

            if (null != activeMQSession) {
                activeMQSession.setMessageConsumer(null); // 设为空，下次重新连接
            }

            return 0;
        }

        // 其他错误
        throw new RuntimeException(e);
    }
}